package com.suxin.socket.yinfei.pusher.sync;

import com.alibaba.fastjson.JSONObject;
import com.suxin.socket.yinfei.session.SessionConnectionHolder;
import lombok.Data;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * @author Tang
 * @classname KafkaSyncPusherHandler
 * @description [  ]
 * @date 2021/8/24 21:52
 */
@Slf4j
public class KafkaSyncPusherHandler extends AbstractSyncPusherHandler<String> {

    @Value("${some.property}")
    private String topic;

    @Setter
    private KafkaTemplate<String, String> kafkaTemplate;

    public KafkaSyncPusherHandler(SessionConnectionHolder sessionConnectionHolder) {
        super(sessionConnectionHolder);
    }

    @Override
    protected void doSendMsg(String msg) {
        kafkaTemplate.send(topic, null, msg);
    }

    @KafkaListener(topics = {"${some.property}"})
    public void listener(ConsumerRecord<String, String> consumerRecord) {
        KafkaRequest kafkaRequest = JSONObject.parseObject(consumerRecord.value(), KafkaRequest.class);
        proxyPusherHandler.push(kafkaRequest.getValue(), kafkaRequest.getIdentifier());
    }

    @Data
    public static class KafkaRequest {
        private Object value;
        private String identifier;
    }

}